package com.smarttest.common.util;

import java.io.File;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.io.FileUtils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.g7s.test.utils.Excel.DataHandlers;

public class RockectMqProduce {
	private DefaultMQProducer producer;
	//构造方法
	public RockectMqProduce(String producerGroup,String namesrvAddr,String instanceName) {
		DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
		producer.setNamesrvAddr(namesrvAddr);
		producer.setInstanceName(instanceName);
		this.producer = producer;
	}  
    /**
     * 启动生产者
     */
	public void start() {
		try {
			this.producer.start();
		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	/**
	 * 发送MQ消息
	 * @param msglist
	 * @param topic
	 * @param tag
	 * @param key
	 * @param sendtime
	 * @param timestamp
	 * @param internal
	 * @param startrownum
	 *            从第几行开始发。为0、负数或超过最大条数时默认从开头发
	 * @param count
	 *            发多少条。为0、负数或超过最大条数时默认发送全部
	 * @return
	 * @throws Exception
	 */
	@SuppressWarnings("static-access")
	public Map<Integer, Date> sendMsg(List<String> msglist, String topic, String tag, String key,String sendtime, String timestamp,int internal,int startrownum, int count, int sendinternal) throws Exception{
		Map<Integer, Date> m = new HashMap<Integer, Date>();
		SimpleDateFormat format = new SimpleDateFormat(timestamp);
		Calendar calender = Calendar.getInstance();
		calender.setTime(format.parse(sendtime));
		String timeLongReg = "<[tT]+[Ii]+[mM]+[eE]+[Ll]+[Oo]+[Nn]+[Gg]+>";
		String timeStringReg = "<[tT]+[Ii]+[mM]+[eE]+[sS]+[tT]+[Rr]+[Ii]+[Nn]+[Gg]+>";
		int startnum = 1;
		if (startrownum > 0 && startrownum <= msglist.size()) {
			startnum = startrownum;
		}
		for (int i = startnum-1; i < msglist.size(); i++) {
			String dateString = format.format(calender.getTime());
			Long dateLong = calender.getTime().getTime();
			Date msgtime = calender.getTime();
			Message msg = null;
			String data = msglist.get(i).replaceAll(timeStringReg, dateString).replaceAll(timeLongReg, String.valueOf(dateLong));
			String jsonStr = JSON.parse(data).toString();
			System.out.println(jsonStr);
			msg = new Message(topic,// topic
					tag,// tag
					key,// key
					jsonStr.getBytes());// body
			System.out.println(msg);
			sendMsg(msg,key);
			m.put(i + 1, msgtime);
			if (internal > 0) {
				calender.add(Calendar.SECOND, +internal);
			}
			if (sendinternal > 0 && i < msglist.size() - 1) {
				Thread.currentThread().sleep(sendinternal*1000);
			}
			if (count <= 0 || count >= msglist.size() - startnum) {
				if (i == msglist.size() - 1) {
					break;
				}
			} else {
				if (i == (startnum - 1) + count - 1) {
					break;
				}
			}
		}
		return m;
	}
	/**
	 * 发送消息
	 * @param msglist
	 * @param topic
	 * @param tag
	 * @param key
	 * @param sendinternal
	 * @throws Exception
	 */
	public void sendMsg(List<String> msglist, String topic, String tag, String key, int sendinternal) throws Exception{
		for (int i = 0; i < msglist.size(); i++) {
			Message msg = null;
			String jsonStr = JSON.parse(msglist.get(i)).toString();
			System.out.println(jsonStr);
			msg = new Message(topic,// topic
					tag,// tag
					key,// key
					jsonStr.getBytes());// body
			System.out.println(msg);
			sendMsg(msg,key);
			if (sendinternal > 0 && i < msglist.size() - 1) {
				Thread.currentThread().sleep(sendinternal*1000);
			}
		}
		
	}
	/**
	 * 通过excel数据发送消息(txt可以转为excel格式)
	 * @param excelFile excel文件名
	 * @param sheetname sheet名
	 * @param topic 要发送消息的topic
	 * @param tag 要发送消息的tag
	 * @param count 要发送消息的条数 0或负数发送全部
	 * @param key 要发送消息的key
	 * @param timeInterval 每次消息发送时间间隔  单位：秒
	 * @return 
	 * @throws Exception 
	 */
	public Map<Integer, Date> sendMsgForExcel (String excelFile, String sheetname,String topic, String tag, String key, String sendtime, String timestamp,  int count,int timeInterval,int sendinternal) throws Exception {
		Object [][] data = DataHandlers.getInstance(excelFile).getData(sheetname,0,-1,0,-1,true);
		Map<Integer, Date> m = new HashMap<Integer, Date>();
		SimpleDateFormat format = new SimpleDateFormat(timestamp);
		Calendar calender = Calendar.getInstance();
		calender.setTime(format.parse(sendtime));
		List<String> l = new ArrayList<String>();
		String timeLongReg = "<[tT]+[Ii]+[mM]+[eE]+[Ll]+[Oo]+[Nn]+[Gg]+>";
		String timeStringReg = "<[tT]+[Ii]+[mM]+[eE]+[sS]+[tT]+[Rr]+[Ii]+[Nn]+[Gg]+>";
		for (int i = 2; i < data.length; i++) {
			if (count <= 0 || count >= data.length - 2) {
				if (i > data.length - 1) {
					break;
				}
			} else {
				if (i > count - 1+2) {
					break;
				}
			}
			Map<String,Object> map = new HashMap<String,Object>();
			String jsonStr;
			String dateString = format.format(calender.getTime());
			Long dateLong = calender.getTime().getTime();
			Date msgtime = calender.getTime();
			m.put(i-1, msgtime);
			
			for (int j = 0; j < data[i].length; j++) {	
				Object value = data[i][j].toString().replaceAll(timeStringReg, dateString).replaceAll(timeLongReg, String.valueOf(dateLong));
				String jsonkey = data[0][j].toString();
				String fomart = data[1][j].toString();
				if (value.toString().isEmpty()) {
					value = value.toString();
				} else {
					if (fomart.equalsIgnoreCase("int")) {
						value = Integer.parseInt(value.toString());
					} else if (fomart.equalsIgnoreCase("long")) {
						value = Long.parseLong(value.toString());
					}else if (fomart.equalsIgnoreCase("short")) {
						value = Short.parseShort(value.toString());
					}else if (fomart.equalsIgnoreCase("byte")) {
						value = Byte.parseByte(value.toString());
					}else if (fomart.equalsIgnoreCase("double")) {
						value = Double.parseDouble(value.toString());
					} else if (fomart.equalsIgnoreCase("float")) {
						value = Float.parseFloat(value.toString());
					}else if (fomart.equalsIgnoreCase("boolean")) {
						value = Boolean.parseBoolean(value.toString());
					} else if (fomart.equalsIgnoreCase("jsonobject")) {
						value = JSON.parseObject(value.toString());
					} else if (fomart.equalsIgnoreCase("jsonarray")) {
						value = JSON.parseArray(value.toString());
					} else {
						value = value.toString();
					}
				}
				map.put(jsonkey, value);				
			}
			jsonStr = JSONObject.toJSONString(map);
			if (timeInterval > 0) {
				calender.add(Calendar.SECOND, +timeInterval);
			}
			l.add(jsonStr);	
		}
		sendMsg(l, topic, tag, key,sendinternal);
		return m;
	}
//	
//	/**
//	 * 通过excel数据发送消息(txt可以转为excel格式)
//	 * @param excelFile excel文件名
//	 * @param sheetname sheet名
//	 * @param topic 要发送消息的topic
//	 * @param tag 要发送消息的tag
//	 * @param key 要发送消息的key
//	 */
//	public void sendMsgForExcel (String excelFile, String sheetname,String topic, String tag, String key) {
//		Object [][] keys = DataHandlers.getInstance(excelFile).getData(sheetname,0,1,0,-1,true);
//		Object [][] data = DataHandlers.getInstance(excelFile).getData(sheetname,1,-1,0,-1,true);
//		for (int i = 0; i < data.length; i++) {
//			Map<String,String> map = new HashMap<String,String>();
//			Message msg = null;
//			String jsonStr;
//			for (int j = 0; j < keys[0].length; j++) {			
//				map.put(keys[0][j].toString(), data[i][j].toString());				
//			}
//			jsonStr = JSONObject.toJSONString(map);
//			System.out.println(jsonStr);
//			msg = new Message(topic,// topic
//					tag,// tag
//					key,// key
//					jsonStr.getBytes());// body
//			System.out.println(msg);
//			try {
//				sendMsg(msg);
//			} catch (Exception e) {
//				// TODO Auto-generated catch block
//				e.printStackTrace();
//				this.producer.shutdown();
//			}
//		}
//	}
	/**
	 * 从文件中读取消息并发送(json格式，通过<timeLong>/<timeString>来设置消息时间)
	 * @param filename 报文存储文件名(.txt/.json/.properties等),文本内容：UTF-8
	 * @param topic 要发送消息的topic
	 * @param tag 要发送消息的tag
	 * @param key 要发送消息的key
	 * @param sendtime
	 *            发送开始时间
	 * @param timestamp
	 *            消息时间格式
	 * @param internal
	 *            每条消息间隔时间(单位:s),如果设置小于等于0，那么该参数无效
	 * @return 
	 * @throws InterruptedException 
	 * @throws MQBrokerException 
	 * @throws RemotingException 
	 * @throws MQClientException 
	 */
	public Map<Integer, Date> sendMsgFromFile(String filename,String topic, String tag, String key,String sendtime, String timestamp,int internal,int sendinternal) throws Exception {
		return sendMsgFromFile(filename, topic, tag, key, sendtime, timestamp, internal, 0, 0,sendinternal);
	}

	/**
	 * 从文件中读取消息并发送(json格式，通过<timeLong>或<timeString>来设置消息时间)
	 * @param filename 报文存储文件名(.txt/.json/.properties等),文本内容：UTF-8
	 * @param topic 要发送消息的topic
	 * @param tag 要发送消息的tag
	 * @param key 要发送消息的key
	 * @param sendtime
	 *            发送开始时间
	 * @param timestamp
	 *            消息时间格式
	 * @param internal
	 *            每条消息间隔时间(单位:s),如果设置小于等于0，那么该参数无效
	 * @param startrownum
	 *            从第几行开始发。为0、负数或超过最大条数时默认从开头发
	 * @param count
	 *            发多少条。为0、负数或超过最大条数时默认发送全部
	 * @return 
	 */
	public Map<Integer, Date> sendMsgFromFile(String filename,String topic, String tag, String key,String sendtime, String timestamp,int internal, int startrownum, int count,int sendinternal) throws Exception {
		List<String> msglist = null;
		SimpleDateFormat format = new SimpleDateFormat(timestamp);
		Calendar calender = Calendar.getInstance();
		calender.setTime(format.parse(sendtime));
		msglist = FileUtils.readLines(new File(filename),"utf-8");
		Iterator<String> i = msglist.iterator();
		while (i.hasNext()) {
			String x = i.next();
			if (x.isEmpty()) {
				i.remove();
			}
		}
		return sendMsg(msglist, topic, tag, key, sendtime, timestamp, internal, startrownum, count,sendinternal);
	}
	/**
	 * 发送json格式的消息
	 * @param topic
	 * @param tag
	 * @param key
	 * @param jsonmsg json格式的消息
	 */
	public void sendMsg(String topic, String tag, String key, JSONObject jsonmsg) {
		Message msg = new Message(topic,// topic
				tag,// tag
				key,// key
				jsonmsg.toString().getBytes());// body
		System.out.println(msg);
		try {
			sendMsg(msg,key);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			this.producer.shutdown();
		}
	}
//	/**
//	 * 发送消息
//	 * @param msg
//	 * @return 消息发送状态
//	 * @throws InterruptedException 
//	 * @throws MQBrokerException 
//	 * @throws RemotingException 
//	 * @throws MQClientException 
//	 */
//    public String sendMsg(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { 
//    	String status = "";
//		SendResult sendResult = this.producer.send(msg);
//		status = sendResult.getSendStatus().toString();				
//        return status;
//    }
    public String sendMsg(Message msg, String key) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { 
    	String status = "";
		SendResult sendResult = this.producer.send(msg, new SelectMessageQueueByHash(), key);
		status = sendResult.getSendStatus().toString();				
        return status;
    }
    /**
     * 关闭生产者
     */
    public void close () {
    	this.producer.shutdown();
    }
}
